package cn.tianyigps.m7.template;

import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;

import java.io.IOException;
import java.util.Properties;

/**
 * Created with IDEA
 *
 * @description: kafka模板类
 * @author: cui.guibin
 * @create: 2019/4/3 16:41
 **/
public class KafkaTemplate {

    private static  Properties props = new Properties();
    private static  KafkaProducer<String, String> producer;
    static {
        try {
            props.load(KafkaTemplate.class.getClassLoader().getResourceAsStream("kafka.properties"));
        } catch (IOException e) {
            e.printStackTrace();
        }
//        props.put("bootstrap.servers", "172.19.25.119:9092");
//        props.put("acks", "all");
//        props.put("retries", 0);
//        props.put("batch.size", 16384);
//        props.put("key.serializer", StringSerializer.class.getName());
//        props.put("value.serializer", StringSerializer.class.getName());
//        props.put("group.id", "m7-group");
//        props.put("enable.auto.commit", "true");
//        props.put("auto.commit.interval.ms", "1000");
//        props.put("session.timeout.ms", "30000");
//        props.put("max.poll.records", 3);
//        props.put("auto.offset.reset", "earliest");
//        props.put("key.deserializer", StringDeserializer.class.getName());
//        props.put("value.deserializer", StringDeserializer.class.getName());

    }
    public static KafkaProducer<String, String> producer(){
        producer = new KafkaProducer<String, String>(props);
        return producer;
    }
    public static KafkaConsumer<String, String> consumer(){
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
        return consumer;
    }

//    public static void main(String[] args) {
//        KafkaConsumer<String, String> consumer = consumer();
//        consumer.subscribe(Arrays.asList("inbound"));
//        for (;;){
//            ConsumerRecords<String, String> consumerRecords = consumer.poll(1);
//            ObjectMapper objectMapper = new ObjectMapper();
//            JavaType javaType = getCollectionType(ArrayList.class, StandardModel.class);
//            if (null != consumerRecords && consumerRecords.count()>0){
//                for (ConsumerRecord consumerRecord:consumerRecords) {
//                    List<StandardModel> standardModels = null;
//                    String value = (String) consumerRecord.value();
//                    try {
//                        standardModels = objectMapper.readValue(value,javaType);
//                    } catch (IOException e) {
//                        e.printStackTrace();
//                    }
//                    System.out.println(value);
//
//                }
//            }
//
//        }
//    }
//    public static JavaType getCollectionType(Class<?> collectionClass, Class<?>... elementClasses) {
//        ObjectMapper objectMapper = new ObjectMapper();
//        return objectMapper.getTypeFactory().constructParametricType(collectionClass, elementClasses);
//    }
}
